package mq

import (
	"context"
	"fmt"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/apache/rocketmq-client-go/v2/producer"
)

type Producer struct {
	nameServer string
	topic      string
	producer   *rocketmq.Producer
}

func NewProducer(server, topic string) *Producer {
	p := &Producer{
		nameServer: server,
		topic:      topic,
		producer:   nil,
	}
	p.newProducer()
	return p
}

func (this *Producer) newProducer() bool {
	//连接recketmq
	p, err := rocketmq.NewProducer(producer.WithNameServer([]string{this.nameServer}))
	if err != nil {
		fmt.Println("NewProducer失败：", err)
		return false
	}
	//启动
	err = p.Start()
	if err != nil {
		fmt.Println("启动producer错误：", err)
		return false
	}
	this.producer = &p
	return true
}

func (this *Producer) Send(json string) error {
	//实例化消息
	msg := &primitive.Message{
		Topic: this.topic,
		Body:  []byte(json),
	}
	//同步发送
	res, err := (*this.producer).SendSync(context.Background(), msg)
	if err != nil {
		fmt.Printf("send message error: %s\n", err)
	} else {
		fmt.Printf("send message success: result=%s\n", res.String())
	}
	return nil
}

func (this *Producer) Shutdown() error {
	if this.producer != nil {
		//关闭连接
		err := (*this.producer).Shutdown()
		if err != nil {
			fmt.Printf("shutdown producer error: %s", err.Error())
			return err
		}
	}
	return nil
}
